package bigdata.backend.meta.service;

import bigdata.backend.datasource.entity.Datasource;
import bigdata.backend.datasource.mapper.DatasourceMapper;
import bigdata.backend.meta.entity.DorisColumnMeta;
import bigdata.backend.meta.entity.DorisTableMeta;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Doris 建表 DDL 生成器（基于持久化的元数据）
 * <p>
 * 职责：
 * - 接收 DorisTableMeta 和 DorisColumnMeta 列表；
 * - 反向生成符合 Apache Doris 2.1.11 语法的 CREATE TABLE 语句；
 * - 支持所有主流模型（DUPLICATE/UNIQUE/AGGREGATE/PRIMARY KEY）；
 * - 支持 RANGE 分区、HASH 分桶、默认值、列/表注释、PROPERTIES；
 * - 安全转义注释中的单引号，防止 SQL 注入或语法错误。
 * </p>
 * <p>
 * 使用说明：
 * - 调用前确保 columns 按 ordinalPosition 排序；
 * - partitionInfo 需包含 column 和 values 字段；
 * - keyColumns、bucketColumns、properties 必须为合法 JSON 字符串。
 * </p>
 */
@Slf4j
@Service
public class DorisDdlGenerator {

    @Autowired
    private DatasourceMapper datasourceMapper;

    private static final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 根据 Doris 表元数据生成完整的 CREATE TABLE DDL 语句
     *
     * @param tableMeta Doris 表主元数据（必填）
     * @param columns   列元数据列表（必填，建议按 ordinalPosition 排序）
     * @return 可直接在 Doris 中执行的建表 SQL
     * @throws IllegalArgumentException 当元数据格式非法或缺失关键信息时抛出
     */
    public String generateDdl(DorisTableMeta tableMeta, List<DorisColumnMeta> columns) {
        if (tableMeta == null) {
            throw new IllegalArgumentException("DorisTableMeta 不能为空");
        }
        if (columns == null || columns.isEmpty()) {
            throw new IllegalArgumentException("字段列表不能为空");
        }

        Datasource datasource = datasourceMapper.selectById(tableMeta.getDatasourceId());

        log.debug("开始为 Doris 表 [{}.{}] 生成 DDL", datasource.getDefaultDatabase(), tableMeta.getTableName());

        StringBuilder ddl = new StringBuilder();

        // 1. 建表语句头
        ddl.append("-- Generated by Meta System for Apache Doris 2.1.11\n");
        ddl.append("-- Table: ").append(tableMeta.getTableName()).append("\n");
        ddl.append("-- Created at: ").append(LocalDateTime.now()).append("\n");
        ddl.append("CREATE TABLE IF NOT EXISTS `")
                .append(datasource.getDefaultDatabase()).append("`.`")
                .append(tableMeta.getTableName()).append("` (\n");

        // 2. 列定义
        List<String> columnDefs = new ArrayList<>();
        for (DorisColumnMeta col : columns) {
            String def = buildColumnDefinition(col);
            columnDefs.add("  " + def); // 缩进 2 空格，提升可读性
        }
        ddl.append(String.join(",\n", columnDefs));
        ddl.append("\n) ");

        // 3. KEY 模型定义（如 UNIQUE KEY(id)）
        String keyClause = buildKeyClause(tableMeta);
        ddl.append(keyClause).append("\n");

        // 4. 表注释（Doris 2.0+ 支持）
        if (tableMeta.getTableComment() != null && !tableMeta.getTableComment().trim().isEmpty()) {
            // Doris 使用双引号或无引号字符串作为 COMMENT 值，但为安全起见，使用双引号并转义
            String safeComment = escapeDoubleQuote(tableMeta.getTableComment().trim());
            ddl.append("COMMENT \"").append(safeComment).append("\"\n");
        }

        // 5. 分区定义（PARTITION BY RANGE）
        String partitionClause = buildPartitionClause(tableMeta);
        if (partitionClause != null && !partitionClause.trim().isEmpty()) {
            ddl.append(partitionClause).append("\n");
        }

        // 6. 分桶定义（DISTRIBUTED BY HASH）
        String distributeClause = buildDistributeClause(tableMeta);
        ddl.append(distributeClause).append("\n");

        // 7. PROPERTIES 属性
        String propertiesClause = buildPropertiesClause(tableMeta);
        ddl.append(propertiesClause).append(";");

        log.debug("DDL 生成完成: \n{}", ddl.toString());
        return ddl.toString();
    }

    /**
     * 构建单个列的定义语句
     * <p>
     * 格式示例：
     * `id` BIGINT NOT NULL COMMENT '主键'
     * `name` VARCHAR(64) NULL DEFAULT "N" COMMENT '姓名'
     * `score` DOUBLE SUM COMMENT '聚合分数'
     * </p>
     */
    public String buildColumnDefinition(DorisColumnMeta col) {
        StringBuilder sb = new StringBuilder();

        // 列名（用反引号包裹，防止关键字冲突）
        sb.append("`").append(col.getColumnName()).append("` ");

        // 智能清理类型
        String cleanType = normalizeDorisColumnType(col.getColumnType());
        sb.append(cleanType);

        // 聚合模型类型（仅 AGGREGATE KEY 表需要，如 SUM, MAX）
        if (col.getAggregationType() != null && !col.getAggregationType().trim().isEmpty()) {
            sb.append(" ").append(col.getAggregationType().trim());
        }

        // NULL / NOT NULL
        // Doris 默认允许 NULL，所以只有非空时才显式声明 NOT NULL
        if (Boolean.FALSE.equals(col.getNullable())) {
            sb.append(" NOT NULL");
        } else {
            sb.append(" NULL");
        }

        // 默认值（Doris 仅支持常量，默认值需严格按类型处理）
        if (col.getDefaultValue() != null && !col.getDefaultValue().trim().isEmpty()) {
            String defaultValue = col.getDefaultValue().trim();
            String columnType = col.getColumnType(); // 如 "VARCHAR(64)", "BIGINT", "DATETIME"

            if ("NULL".equalsIgnoreCase(defaultValue)) {
                sb.append(" DEFAULT NULL");
            } else if (isDorisStringType(columnType)) {
                // 字符串类型：用双引号包裹，转义内部双引号
                sb.append(" DEFAULT \"").append(escapeDoubleQuote(defaultValue)).append("\"");
            } else if (isNumericConstant(defaultValue)) {
                // 数值常量：直接写，不加引号
                sb.append(" DEFAULT ").append(defaultValue);
            } else if ("CURRENT_TIMESTAMP".equalsIgnoreCase(defaultValue.replace("()", ""))) {
                // 谨慎支持 CURRENT_TIMESTAMP（仅当用户明确指定且列为 DATETIME/DATE）
                // 但 Doris 大多数场景不支持，建议日志警告或丢弃
                log.warn("Doris 通常不支持 DEFAULT CURRENT_TIMESTAMP，列: {}, 类型: {}", col.getColumnName(), columnType);
                // 可选择丢弃或保留（按需）
                sb.append(" DEFAULT CURRENT_TIMESTAMP");
            } else {
                // 未知/非法默认值：记录警告并跳过（避免建表失败）
                log.warn("跳过非法默认值 [{}] for column [{}], type [{}]", defaultValue, col.getColumnName(), columnType);
                // 不添加 DEFAULT 子句
            }
        }

        // 列注释
        if (col.getColumnComment() != null && !col.getColumnComment().trim().isEmpty()) {
            String safeComment = escapeSingleQuote(col.getColumnComment().trim());
            sb.append(" COMMENT \"").append(safeComment).append("\"");
        }

        return sb.toString();
    }

    private boolean isDorisStringType(String columnType) {
        if (columnType == null) return false;
        String lower = columnType.toLowerCase();
        return lower.contains("char") || lower.contains("varchar") || lower.contains("string") || lower.contains("text");
    }

    private boolean isNumericConstant(String value) {
        if (value == null || value.isEmpty()) return false;
        // 允许整数、小数、负数
        return value.matches("^-?\\d+(\\.\\d+)?$");
    }

    /**
     * 构建 KEY 模型子句
     * <p>
     * 支持：
     * - DUPLICATE KEY(col1, col2)
     * - UNIQUE KEY(col1)
     * - AGGREGATE KEY(col1)
     * - PRIMARY KEY(col1, col2)
     * </p>
     */
    private String buildKeyClause(DorisTableMeta tableMeta) {
        try {
            // 反序列化 keyColumns JSON 字符串 -> List<String>
            List<String> keyCols = objectMapper.readValue(tableMeta.getKeyColumns(), new TypeReference<List<String>>() {});
            if (keyCols == null || keyCols.isEmpty()) {
                throw new IllegalArgumentException("Key 列不能为空");
            }

            String keyType = tableMeta.getTableType(); // 如 "UNIQUE", "AGGREGATE" 等
            if (keyType == null || keyType.trim().isEmpty()) {
                throw new IllegalArgumentException("表模型类型（tableType）不能为空");
            }

            // ====== 标准化 keyType ======
            String normalizedKeyType = keyType.toUpperCase().trim();
            if (normalizedKeyType.endsWith("_KEY")) {
                normalizedKeyType = normalizedKeyType.substring(0, normalizedKeyType.length() - 4); // 去掉 "_KEY"
            }

            String keyColsStr = keyCols.stream()
                    .map(col -> "`" + col + "`")
                    .collect(Collectors.joining(", "));

            switch (normalizedKeyType) {
                case "DUPLICATE":
                    return "DUPLICATE KEY(" + keyColsStr + ")";
                case "UNIQUE":
                    return "UNIQUE KEY(" + keyColsStr + ")";
                case "AGGREGATE":
                    return "AGGREGATE KEY(" + keyColsStr + ")";
                case "PRIMARY":
                    return "PRIMARY KEY(" + keyColsStr + ")";
                default:
                    throw new IllegalArgumentException("不支持的 Doris 表模型类型: " + keyType);
            }
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("解析 keyColumns JSON 失败: " + tableMeta.getKeyColumns(), e);
        }
    }

    /**
     * 构建 RANGE 分区子句
     * <p>
     * 要求 partitionInfo 格式为：
     * {"column": "dt", "values": ["2025-01-01", "2025-02-01"]}
     * 生成：
     * PARTITION BY RANGE(`dt`) (
     *   PARTITION p202501 VALUES LESS THAN ("2025-01-01"),
     *   PARTITION p202502 VALUES LESS THAN ("2025-02-01"),
     *   PARTITION pmax VALUES LESS THAN MAXVALUE
     * )
     * </p>
     */
    private String buildPartitionClause(DorisTableMeta tableMeta) {
        if (tableMeta.getPartitionInfo() == null || tableMeta.getPartitionInfo().trim().isEmpty()) {
            return null;
        }

        try {
            Map<String, Object> partitionInfo = objectMapper.readValue(tableMeta.getPartitionInfo(), Map.class);
            if (partitionInfo.isEmpty()) {
                return null;
            }

            String partitionColumn = (String) partitionInfo.get("column");
            List<String> partitionValues = (List<String>) partitionInfo.get("values");

            if (partitionColumn == null || partitionValues == null || partitionValues.isEmpty()) {
                return null;
            }

            StringBuilder sb = new StringBuilder("PARTITION BY RANGE(`").append(partitionColumn).append("`) (\n");

            List<String> partitionDefs = new ArrayList<>();
            for (int i = 0; i < partitionValues.size(); i++) {
                String value = partitionValues.get(i).trim();
                // 生成分区名：p + 日期（去掉横杠）
                String partName = "p" + value.replace("-", "");
                // 最后一个分区用 MAXVALUE
                String lessThan = (i == partitionValues.size() - 1)
                        ? "MAXVALUE"
                        : "\"" + escapeDoubleQuote(value) + "\"";

                partitionDefs.add("  PARTITION " + partName + " VALUES LESS THAN (" + lessThan + ")");
            }

            sb.append(String.join(",\n", partitionDefs));
            sb.append("\n)");

            return sb.toString();
        } catch (Exception e) {
            throw new IllegalArgumentException("解析 partitionInfo 失败: " + tableMeta.getPartitionInfo(), e);
        }
    }

    /**
     * 构建分桶（DISTRIBUTED BY）子句
     */
    private String buildDistributeClause(DorisTableMeta tableMeta) {
        try {
            List<String> bucketCols = objectMapper.readValue(tableMeta.getBucketColumns(), new TypeReference<List<String>>() {});
            if (bucketCols == null || bucketCols.isEmpty()) {
                // 如果未指定分桶列，默认使用 keyColumns
                // 但为安全起见，这里要求必须有 bucketColumns（你的 meta 应已存储）
                throw new IllegalArgumentException("分桶列不能为空");
            }

            String cols = bucketCols.stream()
                    .map(col -> "`" + col + "`")
                    .collect(Collectors.joining(", "));

            Integer bucketNum = tableMeta.getBucketNum();
            if (bucketNum == null || bucketNum <= 0) {
                bucketNum = 10; // 默认 10 个分桶
            }

            return "DISTRIBUTED BY HASH(" + cols + ") BUCKETS " + bucketNum;
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("解析 bucketColumns JSON 失败: " + tableMeta.getBucketColumns(), e);
        }
    }

    /**
     * 构建 PROPERTIES 子句
     * <p>
     * 示例：
     * PROPERTIES("replication_num" = "3", "enable_persistent_index" = "true")
     * </p>
     */
    private String buildPropertiesClause(DorisTableMeta tableMeta) {
        try {
            Map<String, String> props = new HashMap<>();

            // 1. 先加载原始 properties（来自 SHOW CREATE TABLE）
            if (tableMeta.getProperties() != null && !tableMeta.getProperties().trim().isEmpty()) {
                Map<String, String> storedProps = objectMapper.readValue(
                        tableMeta.getProperties(),
                        new TypeReference<Map<String, String>>() {}
                );
                props.putAll(storedProps);
            }

            // 2. 【关键修复】只有当没有 replication_allocation 时，才补充 replication_num
            if (!props.containsKey("replication_allocation") && !props.containsKey("replication_num")) {
                // 两者都没有，才设默认 replication_num
                props.put("replication_num", "1");
            }
            // 如果已有 replication_allocation，则完全跳过 replication_num

            // 3. 拼接 PROPERTIES
            String propsStr = props.entrySet().stream()
                    .map(entry -> {
                        String key = escapeDoubleQuote(entry.getKey());
                        String value = escapeDoubleQuote(entry.getValue());
                        return "\"" + key + "\" = \"" + value + "\"";
                    })
                    .collect(Collectors.joining(", "));

            return "PROPERTIES(" + propsStr + ")";
        } catch (Exception e) {
            throw new IllegalArgumentException("解析 properties JSON 失败: " + tableMeta.getProperties(), e);
        }
    }

    /**
     * 转义双引号：用于 DEFAULT 和 PROPERTIES 中的字符串
     * Doris 中双引号字符串内双引号需写成 ""
     */
    private String escapeDoubleQuote(String input) {
        if (input == null) return "";
        return input.replace("\"", "\"\"");
    }

    /**
     * 转义单引号：用于 COMMENT 中（Doris 的 COMMENT 使用单引号）
     * 单引号内单引号需写成 ''
     */
    private String escapeSingleQuote(String input) {
        if (input == null) return "";
        return input.replace("'", "''");
    }

    /**
     * 根据 Doris 语义清理 COLUMN_TYPE 字符串
     *
     * 背景：Doris 的 information_schema.COLUMNS.COLUMN_TYPE 可能返回：
     *   - bigint(20)      → 应简化为 bigint
     *   - int(11)         → 应简化为 int
     *   - varchar(64)     → 必须保留
     *   - decimal(16,4)   → 必须保留
     *
     * 规则：
     *   1. 整数类型（bigint/int/smallint/tinyint）移除括号；
     *   2. 其他类型原样保留。
     */
    private String normalizeDorisColumnType(String columnType) {
        if (columnType == null || columnType.isEmpty()) {
            return columnType;
        }

        String lower = columnType.toLowerCase();

        // 匹配：bigint(20), int(11), smallint(6), tinyint(4)
        if (lower.matches("^(bigint|int|integer|smallint|tinyint)\\(\\d+\\)$")) {
            // 提取基础类型名（如 "bigint"）
            return lower.substring(0, lower.indexOf('('));
        }

        // 其他类型（varchar, char, decimal, string, date 等）原样返回
        return columnType;
    }
}